// Copyright (c) 2011 The LevelDB Authors. All rights reserved.
// Use of this source code is governed by a BSD-style license that can be
// found in the LICENSE file. See the AUTHORS file for names of contributors.

#include "leveldb/table.h"

#include <map>
#include <string>

#include "gtest/gtest.h"
#include "db/dbformat.h"
#include "db/memtable.h"
#include "db/write_batch_internal.h"
#include "leveldb/db.h"
#include "leveldb/env.h"
#include "leveldb/iterator.h"
#include "leveldb/table_builder.h"
#include "table/block.h"
#include "table/block_builder.h"
#include "table/format.h"
#include "util/random.h"
#include "util/testutil.h"

namespace leveldb {

// Return reverse of "key".
// Used to test non-lexicographic comparators.
    static std::string Reverse(const Slice &key) {
        std::string str(key.ToString());
        std::string rev("");
        for (std::string::reverse_iterator rit = str.rbegin(); rit != str.rend();
             ++rit) {
            rev.push_back(*rit);
        }
        return rev;
    }

    namespace {
        class ReverseKeyComparator : public Comparator {
        public:
            const char *Name() const override {
                return "leveldb.ReverseBytewiseComparator";
            }

            int Compare(const Slice &a, const Slice &b) const override {
                return BytewiseComparator()->Compare(Reverse(a), Reverse(b));
            }

            void FindShortestSeparator(std::string *start,
                                       const Slice &limit) const override {
                std::string s = Reverse(*start);
                std::string l = Reverse(limit);
                BytewiseComparator()->FindShortestSeparator(&s, l);
                *start = Reverse(s);
            }

            void FindShortSuccessor(std::string *key) const override {
                std::string s = Reverse(*key);
                BytewiseComparator()->FindShortSuccessor(&s);
                *key = Reverse(s);
            }
        };
    }  // namespace
    static ReverseKeyComparator reverse_key_comparator;

    static void Increment(const Comparator *cmp, std::string *key) {
        if (cmp == BytewiseComparator()) {
            key->push_back('\0');
        } else {
            assert(cmp == &reverse_key_comparator);
            std::string rev = Reverse(*key);
            rev.push_back('\0');
            *key = Reverse(rev);
        }
    }

// An STL comparator that uses a Comparator
    namespace {
        struct STLLessThan {
            const Comparator *cmp;

            STLLessThan() : cmp(BytewiseComparator()) {}

            STLLessThan(const Comparator *c) : cmp(c) {}

            bool operator()(const std::string &a, const std::string &b) const {
                return cmp->Compare(Slice(a), Slice(b)) < 0;
            }
        };
    }  // namespace

    class StringSink : public WritableFile {
    public:
        ~StringSink() override = default;

        const std::string &contents() const { return contents_; }

        Status Close() override { return Status::OK(); }

        Status Flush() override { return Status::OK(); }

        Status Sync() override { return Status::OK(); }

        Status Append(const Slice &data) override {
            contents_.append(data.data(), data.size());
            return Status::OK();
        }

    private:
        std::string contents_;
    };

    class StringSource : public RandomAccessFile {
    public:
        StringSource(const Slice &contents)
                : contents_(contents.data(), contents.size()) {}

        ~StringSource() override = default;

        uint64_t Size() const { return contents_.size(); }

        Status Read(uint64_t offset, size_t n, Slice *result,
                    char *scratch) const override {
            if (offset >= contents_.size()) {
                return Status::InvalidArgument("invalid Read offset");
            }
            if (offset + n > contents_.size()) {
                n = contents_.size() - offset;
            }
            std::memcpy(scratch, &contents_[offset], n);
            *result = Slice(scratch, n);
            return Status::OK();
        }

    private:
        std::string contents_;
    };

    typedef std::map<std::string, std::string, STLLessThan> KVMap;

// Helper class for tests to unify the interface between
// BlockBuilder/TableBuilder and Block/Table.
    class Constructor {
    public:
        explicit Constructor(const Comparator *cmp) : data_(STLLessThan(cmp)) {}

        virtual ~Constructor() = default;

        void Add(const std::string &key, const Slice &value) {
            data_[key] = value.ToString();
        }

        // Finish constructing the data structure with all the keys that have
        // been added so far.  Returns the keys in sorted order in "*keys"
        // and stores the key/value pairs in "*kvmap"
        void Finish(const Options &options, std::vector<std::string> *keys,
                    KVMap *kvmap) {
            *kvmap = data_;
            keys->clear();
            for (const auto &kvp : data_) {
                keys->push_back(kvp.first);
            }
            data_.clear();
            Status s = FinishImpl(options, *kvmap);
            ASSERT_TRUE(s.ok()) << s.ToString();
        }

        // Construct the data structure from the data in "data"
        virtual Status FinishImpl(const Options &options, const KVMap &data) = 0;

        virtual Iterator *NewIterator() const = 0;

        const KVMap &data() const { return data_; }

        virtual DB *db() const { return nullptr; }  // Overridden in DBConstructor

    private:
        KVMap data_;
    };

    class BlockConstructor : public Constructor {
    public:
        explicit BlockConstructor(const Comparator *cmp)
                : Constructor(cmp), comparator_(cmp), block_(nullptr) {}

        ~BlockConstructor() override { delete block_; }

        Status FinishImpl(const Options &options, const KVMap &data) override {
            delete block_;
            block_ = nullptr;
            BlockBuilder builder(&options);

            for (const auto &kvp : data) {
                builder.Add(kvp.first, kvp.second);
            }
            // Open the block
            data_ = builder.Finish().ToString();
            BlockContents contents;
            contents.data = data_;
            contents.cachable = false;
            contents.heap_allocated = false;
            block_ = new Block(contents);
            return Status::OK();
        }

        Iterator *NewIterator() const override {
            return block_->NewIterator(comparator_);
        }

    private:
        const Comparator *const comparator_;
        std::string data_;
        Block *block_;

        BlockConstructor();
    };

    class TableConstructor : public Constructor {
    public:
        TableConstructor(const Comparator *cmp)
                : Constructor(cmp), source_(nullptr), table_(nullptr) {}

        ~TableConstructor() override { Reset(); }

        Status FinishImpl(const Options &options, const KVMap &data) override {
            Reset();
            StringSink sink;
            TableBuilder builder(options, &sink);

            for (const auto &kvp : data) {
                builder.Add(kvp.first, kvp.second);
                EXPECT_LEVELDB_OK(builder.status());
            }
            Status s = builder.Finish();
            EXPECT_LEVELDB_OK(s);

            EXPECT_EQ(sink.contents().size(), builder.FileSize());

            // Open the table
            source_ = new StringSource(sink.contents());
            Options table_options;
            table_options.comparator = options.comparator;
            return Table::Open(table_options, source_, sink.contents().size(), &table_);
        }

        Iterator *NewIterator() const override {
            return table_->NewIterator(ReadOptions());
        }

        uint64_t ApproximateOffsetOf(const Slice &key) const {
            return table_->ApproximateOffsetOf(key);
        }

    private:
        void Reset() {
            delete table_;
            delete source_;
            table_ = nullptr;
            source_ = nullptr;
        }

        StringSource *source_;
        Table *table_;

        TableConstructor();
    };

// A helper class that converts internal format keys into user keys
    class KeyConvertingIterator : public Iterator {
    public:
        explicit KeyConvertingIterator(Iterator *iter) : iter_(iter) {}

        KeyConvertingIterator(const KeyConvertingIterator &) = delete;

        KeyConvertingIterator &operator=(const KeyConvertingIterator &) = delete;

        ~KeyConvertingIterator() override { delete iter_; }

        bool Valid() const override { return iter_->Valid(); }

        void Seek(const Slice &target) override {
            ParsedInternalKey ikey(target, kMaxSequenceNumber, kTypeValue);
            std::string encoded;
            AppendInternalKey(&encoded, ikey);
            iter_->Seek(encoded);
        }

        void SeekToFirst() override { iter_->SeekToFirst(); }

        void SeekToLast() override { iter_->SeekToLast(); }

        void Next() override { iter_->Next(); }

        void Prev() override { iter_->Prev(); }

        Slice key() const override {
            assert(Valid());
            ParsedInternalKey key;
            if (!ParseInternalKey(iter_->key(), &key)) {
                status_ = Status::Corruption("malformed internal key");
                return Slice("corrupted key");
            }
            return key.user_key;
        }

        Slice value() const override { return iter_->value(); }

        Status status() const override {
            return status_.ok() ? iter_->status() : status_;
        }

    private:
        mutable Status status_;
        Iterator *iter_;
    };

    class MemTableConstructor : public Constructor {
    public:
        explicit MemTableConstructor(const Comparator *cmp)
                : Constructor(cmp), internal_comparator_(cmp) {
            memtable_ = new MemTable(internal_comparator_);
            memtable_->Ref();
        }

        ~MemTableConstructor() override { memtable_->Unref(); }

        Status FinishImpl(const Options &options, const KVMap &data) override {
            memtable_->Unref();
            memtable_ = new MemTable(internal_comparator_);
            memtable_->Ref();
            int seq = 1;
            for (const auto &kvp : data) {
                memtable_->Add(seq, kTypeValue, kvp.first, kvp.second);
                seq++;
            }
            return Status::OK();
        }

        Iterator *NewIterator() const override {
            return new KeyConvertingIterator(memtable_->NewIterator());
        }

    private:
        const InternalKeyComparator internal_comparator_;
        MemTable *memtable_;
    };

    class DBConstructor : public Constructor {
    public:
        explicit DBConstructor(const Comparator *cmp)
                : Constructor(cmp), comparator_(cmp) {
            db_ = nullptr;
            NewDB();
        }

        ~DBConstructor() override { delete db_; }

        Status FinishImpl(const Options &options, const KVMap &data) override {
            delete db_;
            db_ = nullptr;
            NewDB();
            for (const auto &kvp : data) {
                WriteBatch batch;
                batch.Put(kvp.first, kvp.second);
                EXPECT_TRUE(db_->Write(WriteOptions(), &batch).ok());
            }
            return Status::OK();
        }

        Iterator *NewIterator() const override {
            return db_->NewIterator(ReadOptions());
        }

        DB *db() const override { return db_; }

    private:
        void NewDB() {
            std::string name = testing::TempDir() + "table_testdb";

            Options options;
            options.comparator = comparator_;
            Status status = DestroyDB(name, options);
            ASSERT_TRUE(status.ok()) << status.ToString();

            options.create_if_missing = true;
            options.error_if_exists = true;
            options.write_buffer_size = 10000;  // Something small to force merging
            status = DB::Open(options, name, &db_);
            ASSERT_TRUE(status.ok()) << status.ToString();
        }

        const Comparator *const comparator_;
        DB *db_;
    };

    enum TestType {
        TABLE_TEST, BLOCK_TEST, MEMTABLE_TEST, DB_TEST
    };

    struct TestArgs {
        TestType type;
        bool reverse_compare;
        int restart_interval;
    };

    static const TestArgs kTestArgList[] = {
            {TABLE_TEST,    false, 16},
            {TABLE_TEST,    false, 1},
            {TABLE_TEST,    false, 1024},
            {TABLE_TEST,    true,  16},
            {TABLE_TEST,    true,  1},
            {TABLE_TEST,    true,  1024},

            {BLOCK_TEST,    false, 16},
            {BLOCK_TEST,    false, 1},
            {BLOCK_TEST,    false, 1024},
            {BLOCK_TEST,    true,  16},
            {BLOCK_TEST,    true,  1},
            {BLOCK_TEST,    true,  1024},

            // Restart interval does not matter for memtables
            {MEMTABLE_TEST, false, 16},
            {MEMTABLE_TEST, true,  16},

            // Do not bother with restart interval variations for DB
            {DB_TEST,       false, 16},
            {DB_TEST,       true,  16},
    };
    static const int kNumTestArgs = sizeof(kTestArgList) / sizeof(kTestArgList[0]);

    class Harness : public testing::Test {
    public:
        Harness() : constructor_(nullptr) {}

        void Init(const TestArgs &args) {
            delete constructor_;
            constructor_ = nullptr;
            options_ = Options();

            options_.block_restart_interval = args.restart_interval;
            // Use shorter block size for tests to exercise block boundary
            // conditions more.
            options_.block_size = 256;
            if (args.reverse_compare) {
                options_.comparator = &reverse_key_comparator;
            }
            switch (args.type) {
                case TABLE_TEST:
                    constructor_ = new TableConstructor(options_.comparator);
                    break;
                case BLOCK_TEST:
                    constructor_ = new BlockConstructor(options_.comparator);
                    break;
                case MEMTABLE_TEST:
                    constructor_ = new MemTableConstructor(options_.comparator);
                    break;
                case DB_TEST:
                    constructor_ = new DBConstructor(options_.comparator);
                    break;
            }
        }

        ~Harness() { delete constructor_; }

        void Add(const std::string &key, const std::string &value) {
            constructor_->Add(key, value);
        }

        void Test(Random *rnd) {
            std::vector<std::string> keys;
            KVMap data;
            constructor_->Finish(options_, &keys, &data);

            TestForwardScan(keys, data);
            TestBackwardScan(keys, data);
            TestRandomAccess(rnd, keys, data);
        }

        void TestForwardScan(const std::vector<std::string> &keys,
                             const KVMap &data) {
            Iterator *iter = constructor_->NewIterator();
            ASSERT_TRUE(!iter->Valid());
            iter->SeekToFirst();
            for (KVMap::const_iterator model_iter = data.begin();
                 model_iter != data.end(); ++model_iter) {
                ASSERT_EQ(ToString(data, model_iter), ToString(iter));
                iter->Next();
            }
            ASSERT_TRUE(!iter->Valid());
            delete iter;
        }

        void TestBackwardScan(const std::vector<std::string> &keys,
                              const KVMap &data) {
            Iterator *iter = constructor_->NewIterator();
            ASSERT_TRUE(!iter->Valid());
            iter->SeekToLast();
            for (KVMap::const_reverse_iterator model_iter = data.rbegin();
                 model_iter != data.rend(); ++model_iter) {
                ASSERT_EQ(ToString(data, model_iter), ToString(iter));
                iter->Prev();
            }
            ASSERT_TRUE(!iter->Valid());
            delete iter;
        }

        void TestRandomAccess(Random *rnd, const std::vector<std::string> &keys,
                              const KVMap &data) {
            static const bool kVerbose = false;
            Iterator *iter = constructor_->NewIterator();
            ASSERT_TRUE(!iter->Valid());
            KVMap::const_iterator model_iter = data.begin();
            if (kVerbose) std::fprintf(stderr, "---\n");
            for (int i = 0; i < 200; i++) {
                const int toss = rnd->Uniform(5);
                switch (toss) {
                    case 0: {
                        if (iter->Valid()) {
                            if (kVerbose) std::fprintf(stderr, "Next\n");
                            iter->Next();
                            ++model_iter;
                            ASSERT_EQ(ToString(data, model_iter), ToString(iter));
                        }
                        break;
                    }

                    case 1: {
                        if (kVerbose) std::fprintf(stderr, "SeekToFirst\n");
                        iter->SeekToFirst();
                        model_iter = data.begin();
                        ASSERT_EQ(ToString(data, model_iter), ToString(iter));
                        break;
                    }

                    case 2: {
                        std::string key = PickRandomKey(rnd, keys);
                        model_iter = data.lower_bound(key);
                        if (kVerbose)
                            std::fprintf(stderr, "Seek '%s'\n", EscapeString(key).c_str());
                        iter->Seek(Slice(key));
                        ASSERT_EQ(ToString(data, model_iter), ToString(iter));
                        break;
                    }

                    case 3: {
                        if (iter->Valid()) {
                            if (kVerbose) std::fprintf(stderr, "Prev\n");
                            iter->Prev();
                            if (model_iter == data.begin()) {
                                model_iter = data.end();  // Wrap around to invalid value
                            } else {
                                --model_iter;
                            }
                            ASSERT_EQ(ToString(data, model_iter), ToString(iter));
                        }
                        break;
                    }

                    case 4: {
                        if (kVerbose) std::fprintf(stderr, "SeekToLast\n");
                        iter->SeekToLast();
                        if (keys.empty()) {
                            model_iter = data.end();
                        } else {
                            std::string last = data.rbegin()->first;
                            model_iter = data.lower_bound(last);
                        }
                        ASSERT_EQ(ToString(data, model_iter), ToString(iter));
                        break;
                    }
                }
            }
            delete iter;
        }

        std::string ToString(const KVMap &data, const KVMap::const_iterator &it) {
            if (it == data.end()) {
                return "END";
            } else {
                return "'" + it->first + "->" + it->second + "'";
            }
        }

        std::string ToString(const KVMap &data,
                             const KVMap::const_reverse_iterator &it) {
            if (it == data.rend()) {
                return "END";
            } else {
                return "'" + it->first + "->" + it->second + "'";
            }
        }

        std::string ToString(const Iterator *it) {
            if (!it->Valid()) {
                return "END";
            } else {
                return "'" + it->key().ToString() + "->" + it->value().ToString() + "'";
            }
        }

        std::string PickRandomKey(Random *rnd, const std::vector<std::string> &keys) {
            if (keys.empty()) {
                return "foo";
            } else {
                const int index = rnd->Uniform(keys.size());
                std::string result = keys[index];
                switch (rnd->Uniform(3)) {
                    case 0:
                        // Return an existing key
                        break;
                    case 1: {
                        // Attempt to return something smaller than an existing key
                        if (!result.empty() && result[result.size() - 1] > '\0') {
                            result[result.size() - 1]--;
                        }
                        break;
                    }
                    case 2: {
                        // Return something larger than an existing key
                        Increment(options_.comparator, &result);
                        break;
                    }
                }
                return result;
            }
        }

        // Returns nullptr if not running against a DB
        DB *db() const { return constructor_->db(); }

    private:
        Options options_;
        Constructor *constructor_;
    };

// Test empty table/block.
    TEST_F(Harness, Empty
    ) {
    for (
    int i = 0;
    i<kNumTestArgs;
    i++) {
    Init(kTestArgList[i]);
    Random rnd(test::RandomSeed() + 1);
    Test(&rnd);
}
}

// Special test for a block with no restart entries.  The C++ leveldb
// code never generates such blocks, but the Java version of leveldb
// seems to.
TEST_F(Harness, ZeroRestartPointsInBlock
) {
char data[sizeof(uint32_t)];
memset(data,
0, sizeof(data));
BlockContents contents;
contents.
data = Slice(data, sizeof(data));
contents.
cachable = false;
contents.
heap_allocated = false;
Block block(contents);
Iterator *iter = block.NewIterator(BytewiseComparator());
iter->

SeekToFirst();

ASSERT_TRUE(!iter->

Valid()

);
iter->

SeekToLast();

ASSERT_TRUE(!iter->

Valid()

);
iter->Seek("foo");
ASSERT_TRUE(!iter->

Valid()

);
delete
iter;
}

// Test the empty key
TEST_F(Harness, SimpleEmptyKey
) {
for (
int i = 0;
i<kNumTestArgs;
i++) {
Init(kTestArgList[i]);
Random rnd(test::RandomSeed() + 1);
Add("", "v");
Test(&rnd);
}
}

TEST_F(Harness, SimpleSingle
) {
for (
int i = 0;
i<kNumTestArgs;
i++) {
Init(kTestArgList[i]);
Random rnd(test::RandomSeed() + 2);
Add("abc", "v");
Test(&rnd);
}
}

TEST_F(Harness, SimpleMulti
) {
for (
int i = 0;
i<kNumTestArgs;
i++) {
Init(kTestArgList[i]);
Random rnd(test::RandomSeed() + 3);
Add("abc", "v");
Add("abcd", "v");
Add("ac", "v2");
Test(&rnd);
}
}

TEST_F(Harness, SimpleSpecialKey
) {
for (
int i = 0;
i<kNumTestArgs;
i++) {
Init(kTestArgList[i]);
Random rnd(test::RandomSeed() + 4);
Add("\xff\xff", "v3");
Test(&rnd);
}
}

TEST_F(Harness, Randomized
) {
for (
int i = 0;
i<kNumTestArgs;
i++) {
Init(kTestArgList[i]);
Random rnd(test::RandomSeed() + 5);
for (
int num_entries = 0;
num_entries < 2000;
num_entries += (num_entries < 50 ? 1 : 200)) {
if ((num_entries % 10) == 0) {
std::fprintf(stderr, "case %d of %d: num_entries = %d\n", (i + 1),
int(kNumTestArgs), num_entries
);
}
for (
int e = 0;
e<num_entries;
e++) {
std::string v;
Add(test::RandomKey(&rnd, rnd.Skewed(4)),
        test::RandomString(&rnd, rnd.Skewed(5), &v)
.

ToString()

);
}
Test(&rnd);
}
}
}

TEST_F(Harness, RandomizedLongDB
) {
Random rnd(test::RandomSeed());
TestArgs args = {DB_TEST, false, 16};
Init(args);
int num_entries = 100000;
for (
int e = 0;
e<num_entries;
e++) {
std::string v;
Add(test::RandomKey(&rnd, rnd.Skewed(4)),
        test::RandomString(&rnd, rnd.Skewed(5), &v)
.

ToString()

);
}
Test(&rnd);

// We must have created enough data to force merging
int files = 0;
for (
int level = 0;
level<config::kNumLevels;
level++) {
std::string value;
char name[100];
std::snprintf(name,
sizeof(name), "leveldb.num-files-at-level%d", level);

ASSERT_TRUE (db()

->
GetProperty(name, &value
));
files +=
atoi(value
.

c_str()

);
}
ASSERT_GT(files,
0);
}

TEST(MemTableTest, Simple
) {
InternalKeyComparator cmp(BytewiseComparator());
MemTable *memtable = new MemTable(cmp);
memtable->

Ref();

WriteBatch batch;
WriteBatchInternal::SetSequence(&batch,
100);
batch.
Put(std::string("k1"), std::string("v1")
);
batch.
Put(std::string("k2"), std::string("v2")
);
batch.
Put(std::string("k3"), std::string("v3")
);
batch.
Put(std::string("largekey"), std::string("vlarge")
);
ASSERT_TRUE(WriteBatchInternal::InsertInto(&batch, memtable)
.

ok()

);

Iterator *iter = memtable->NewIterator();
iter->

SeekToFirst();

while (iter->

Valid()

) {
std::fprintf(stderr, "key: '%s' -> '%s'\n", iter->

key()

.

ToString()

.

c_str(),
        iter

->

value()

.

ToString()

.

c_str()

);
iter->

Next();

}

delete
iter;
memtable->

Unref();

}

static bool Between(uint64_t val, uint64_t low, uint64_t high) {
    bool result = (val >= low) && (val <= high);
    if (!result) {
        std::fprintf(stderr, "Value %llu is not in range [%llu, %llu]\n",
                     (unsigned long long) (val), (unsigned long long) (low),
                     (unsigned long long) (high));
    }
    return result;
}

TEST(TableTest, ApproximateOffsetOfPlain
) {
TableConstructor c(BytewiseComparator());
c.Add("k01", "hello");
c.Add("k02", "hello2");
c.Add("k03", std::string(10000, 'x'));
c.Add("k04", std::string(200000, 'x'));
c.Add("k05", std::string(300000, 'x'));
c.Add("k06", "hello3");
c.Add("k07", std::string(100000, 'x'));
std::vector<std::string> keys;
KVMap kvmap;
Options options;
options.
block_size = 1024;
options.
compression = kNoCompression;
c.
Finish(options, &keys, &kvmap
);

ASSERT_TRUE(Between(c.ApproximateOffsetOf("abc"), 0, 0)
);
ASSERT_TRUE(Between(c.ApproximateOffsetOf("k01"), 0, 0)
);
ASSERT_TRUE(Between(c.ApproximateOffsetOf("k01a"), 0, 0)
);
ASSERT_TRUE(Between(c.ApproximateOffsetOf("k02"), 0, 0)
);
ASSERT_TRUE(Between(c.ApproximateOffsetOf("k03"), 0, 0)
);
ASSERT_TRUE(Between(c.ApproximateOffsetOf("k04"), 10000, 11000)
);
ASSERT_TRUE(Between(c.ApproximateOffsetOf("k04a"), 210000, 211000)
);
ASSERT_TRUE(Between(c.ApproximateOffsetOf("k05"), 210000, 211000)
);
ASSERT_TRUE(Between(c.ApproximateOffsetOf("k06"), 510000, 511000)
);
ASSERT_TRUE(Between(c.ApproximateOffsetOf("k07"), 510000, 511000)
);
ASSERT_TRUE(Between(c.ApproximateOffsetOf("xyz"), 610000, 612000)
);
}

static bool SnappyCompressionSupported() {
    std::string out;
    Slice in = "aaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa";
    return port::Snappy_Compress(in.data(), in.size(), &out);
}

TEST(TableTest, ApproximateOffsetOfCompressed
) {
if (!

SnappyCompressionSupported()

) {
std::fprintf(stderr, "skipping compression tests\n");
return;
}

Random rnd(301);
TableConstructor c(BytewiseComparator());
std::string tmp;
c.Add("k01", "hello");
c.Add("k02",
test::CompressibleString(&rnd,
0.25, 10000, &tmp));
c.Add("k03", "hello3");
c.Add("k04",
test::CompressibleString(&rnd,
0.25, 10000, &tmp));
std::vector<std::string> keys;
KVMap kvmap;
Options options;
options.
block_size = 1024;
options.
compression = kSnappyCompression;
c.
Finish(options, &keys, &kvmap
);

// Expected upper and lower bounds of space used by compressible strings.
static const int kSlop = 1000;  // Compressor effectiveness varies.
const int expected = 2500;      // 10000 * compression ratio (0.25)
const int min_z = expected - kSlop;
const int max_z = expected + kSlop;

ASSERT_TRUE(Between(c.ApproximateOffsetOf("abc"), 0, kSlop)
);
ASSERT_TRUE(Between(c.ApproximateOffsetOf("k01"), 0, kSlop)
);
ASSERT_TRUE(Between(c.ApproximateOffsetOf("k02"), 0, kSlop)
);
// Have now emitted a large compressible string, so adjust expected offset.
ASSERT_TRUE(Between(c.ApproximateOffsetOf("k03"), min_z, max_z)
);
ASSERT_TRUE(Between(c.ApproximateOffsetOf("k04"), min_z, max_z)
);
// Have now emitted two large compressible strings, so adjust expected offset.
ASSERT_TRUE(Between(c.ApproximateOffsetOf("xyz"), 2 * min_z, 2 * max_z)
);
}

}  // namespace leveldb

int main(int argc, char **argv) {
    testing::InitGoogleTest(&argc, argv);
    return RUN_ALL_TESTS();
}
